package com.djf.djfcs.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.djf.djfcs.config.CanalProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Canal客户端服务
 */
@Slf4j
@Service
public class CanalClientService {
    
    @Autowired
    private CanalConnector canalConnector;
    
    @Autowired
    private CanalProperties canalProperties;
    
    @Autowired(required = false)
    private List<CanalDataListener> dataListeners = new ArrayList<>();
    
    private volatile boolean running = false;
    private Thread consumerThread;
    
    /**
     * 启动Canal客户端
     */
    @PostConstruct
    public void start() {
        log.info("启动Canal客户端服务...");
        running = true;
        
        consumerThread = new Thread(this::consume, "canal-consumer");
        consumerThread.start();
        
        log.info("Canal客户端服务启动成功");
    }
    
    /**
     * 停止Canal客户端
     */
    @PreDestroy
    public void stop() {
        log.info("停止Canal客户端服务...");
        running = false;
        
        if (consumerThread != null) {
            consumerThread.interrupt();
        }
        
        if (canalConnector != null) {
            canalConnector.disconnect();
        }
        
        log.info("Canal客户端服务已停止");
    }
    
    /**
     * 消费Canal数据
     */
    private void consume() {
        try {
            // 连接Canal服务器
            canalConnector.connect();
            // 订阅数据库表
            canalConnector.subscribe(".*\\..*");
            // 回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拉取
            canalConnector.rollback();
            
            log.info("Canal客户端连接成功，开始消费数据...");
            
            while (running) {
                try {
                    // 获取指定数量的数据
                    Message message = canalConnector.getWithoutAck(
                            canalProperties.getBatchSize(), 
                            canalProperties.getTimeout(), 
                            TimeUnit.MILLISECONDS);
                    
                    long batchId =  message.getId();

                    int size = message.getEntries().size();
                    
                    if (batchId == -1 || size == 0) {
                        // 没有数据，休眠一下
                        Thread.sleep(1000);
                        continue;
                    }
                    
                    log.debug("接收到Canal消息，batchId: {}, size: {}", batchId, size);
                    
                    // 处理数据
                    processEntries(message.getEntries());
                    
                    // 确认消息
                    canalConnector.ack(batchId);
                    
                } catch (Exception e) {
                    log.error("处理Canal消息时发生异常", e);
                    // 发生异常时回滚
                    canalConnector.rollback();
                    Thread.sleep(5000);
                }
            }
            
        } catch (Exception e) {
            log.error("Canal客户端运行异常", e);
        } finally {
            if (canalConnector != null) {
                canalConnector.disconnect();
            }
        }
    }
    
    /**
     * 处理Canal条目
     */
    private void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ||
                entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("解析RowChange失败", e);
                continue;
            }
            
            CanalEntry.EventType eventType = rowChange.getEventType();
            
            // 只处理INSERT、UPDATE、DELETE事件
            if (eventType != CanalEntry.EventType.INSERT &&
                eventType != CanalEntry.EventType.UPDATE &&
                eventType != CanalEntry.EventType.DELETE) {
                continue;
            }
            
            String database = entry.getHeader().getSchemaName();
            String table = entry.getHeader().getTableName();
            
            log.debug("处理数据变更: {}.{}, 事件类型: {}", database, table, eventType);
            
            // 处理每一行数据变更
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                CanalDataChangeEvent event = buildChangeEvent(entry, eventType, rowData);
                notifyListeners(event);
            }
        }
    }
    
    /**
     * 构建数据变更事件
     */
    private CanalDataChangeEvent buildChangeEvent(CanalEntry.Entry entry, 
                                                  CanalEntry.EventType eventType, 
                                                  CanalEntry.RowData rowData) {
        CanalDataChangeEvent event = new CanalDataChangeEvent();
        event.setDatabase(entry.getHeader().getSchemaName());
        event.setTable(entry.getHeader().getTableName());
        event.setEventType(eventType.name());
        event.setExecuteTime(entry.getHeader().getExecuteTime());
        event.setLogfileName(entry.getHeader().getLogfileName());
        event.setLogfileOffset(entry.getHeader().getLogfileOffset());
        
        // 处理变更前的数据
        if (rowData.getBeforeColumnsList() != null && !rowData.getBeforeColumnsList().isEmpty()) {
            List<Map<String, Object>> oldData = new ArrayList<>();
            Map<String, Object> oldRow = new HashMap<>();
            for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                oldRow.put(column.getName(), column.getValue());
            }
            oldData.add(oldRow);
            event.setOldData(oldData);
        }
        
        // 处理变更后的数据
        if (rowData.getAfterColumnsList() != null && !rowData.getAfterColumnsList().isEmpty()) {
            List<Map<String, Object>> newData = new ArrayList<>();
            Map<String, Object> newRow = new HashMap<>();
            for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
                newRow.put(column.getName(), column.getValue());
            }
            newData.add(newRow);
            event.setNewData(newData);
        }
        
        return event;
    }
    
    /**
     * 通知监听器
     */
    private void notifyListeners(CanalDataChangeEvent event) {
        for (CanalDataListener listener : dataListeners) {
            try {
                // 检查监听器是否关心这个事件
                if (shouldNotify(listener, event)) {
                    listener.onDataChange(event);
                }
            } catch (Exception e) {
                log.error("通知监听器时发生异常: {}", listener.getClass().getSimpleName(), e);
            }
        }
    }
    
    /**
     * 判断是否应该通知监听器
     */
    private boolean shouldNotify(CanalDataListener listener, CanalDataChangeEvent event) {
        // 检查数据库
        String database = listener.getDatabase();
        if (database != null && !database.equals(event.getDatabase())) {
            return false;
        }
        
        // 检查表名
        String table = listener.getTable();
        if (table != null && !table.equals(event.getTable())) {
            return false;
        }
        
        // 检查事件类型
        String[] eventTypes = listener.getEventTypes();
        if (eventTypes != null && eventTypes.length > 0) {
            boolean found = false;
            for (String eventType : eventTypes) {
                if (eventType.equals(event.getEventType())) {
                    found = true;
                    break;
                }
            }
            if (!found) {
                return false;
            }
        }
        
        return true;
    }
}